package com.atguigu.chapter08;

import com.atguigu.Bean.OrderEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ClassName: Flink06_Login_Project
 * @Description:  恶意登录监控
 * @Author: kele
 * @Date: 2021/4/10 10:50
 *
 * 需求：订单支付实时监控
 *
 * 为了正确控制业务流程，也为了增加用户的支付意愿，网站一般会设置一个支付失效时间，超过一段时间不支付的订单就会被取消。
 * 另外，对于订单的支付，我们还应保证用户支付的正确性，这可以通过第三方支付平台的交易数据来做一个实时对账。
 *
 **/
public class Flink07_Order_State_Project {


    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(2);

        WatermarkStrategy<OrderEvent> wms = WatermarkStrategy
                .<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                    @Override
                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                        return element.getEventTime();
                    }
                });


        env
//                .socketTextStream("hadoop162",8888)
                .readTextFile("in/OrderLog.csv")
                .map(line -> {
                    String[] datas = line.split(",");
                    return new OrderEvent(
                            Long.valueOf(datas[0]),
                            datas[1],
                            datas[2],
                            Long.parseLong(datas[3]) * 1000);

                })
                .assignTimestampsAndWatermarks(wms)
                .keyBy(OrderEvent::getOrderId)
                .process(new KeyedProcessFunction<Long, OrderEvent, String>() {

                    private ValueState<Long> timerState;
                    private ValueState<OrderEvent> createState;
                    private ValueState<OrderEvent> payState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //定义两个状态
                        //创建订单的状态
                        createState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("createState", OrderEvent.class));

                        //支付的状态
                        payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("payState", OrderEvent.class));

                        timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timerState", Long.class));

                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                        if (createState.value() != null) {

                            out.collect("订单: " + createState.value().getOrderId() + " 有创建, 没有支付");
                            System.out.println(createState.value().getOrderId());

                        } else {

                            out.collect("订单: " + payState.value().getOrderId() + " 有支付, 没有创建, 系统问题");

                        }

                        payState.clear();
                        createState.clear();
                        timerState.clear();
                    }

                    @Override
                    public void processElement(OrderEvent value, Context ctx, Collector<String> out) throws Exception {

                        //第一次到达数据，创建定时器
                        if(timerState.value() == null){

                            timerState.update(value.getEventTime()+ 55 * 60 * 1000);

                            ctx.timerService().registerEventTimeTimer(timerState.value());

                        }
                        //如果两条数据都到了，就取消定时器
                        else {

                            ctx.timerService().deleteEventTimeTimer(timerState.value());

                            timerState.clear();

                        }

                        if("create".equalsIgnoreCase(value.getEventType())){

                            if(payState.value() == null){

                                createState.update(value);

                            }
                            else if( (payState.value().getEventTime() -  value.getEventTime()) <= 45 * 60 * 1000){

                                out.collect(ctx.getCurrentKey() + " 正常创建和支付...");
                                payState.clear();

                            }else{

                                out.collect(ctx.getCurrentKey() + " 创建和支付时间超过45分钟, 检测是否系统问题.");
                                payState.clear();
                            }

                        }else{

                            if(createState.value() == null){
                                payState.update(value);
                            }

                            else if((value.getEventTime() - createState.value().getEventTime()) <= 45 * 60 * 1000){

                                out.collect(ctx.getCurrentKey() + " 正常创建和支付...");
                                createState.clear();

                            }
                            else{
                                out.collect(ctx.getCurrentKey() + " 创建和支付时间超过45分钟, 检测是否系统问题.");
                                createState.clear();
                            }
                        }

                    }
                })
                .print();




        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
